package com.aotain.kafka;

import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;

public abstract class AbstractHanldMsgThread implements Runnable{
	private KafkaStream<byte[], byte[]> kafkaStream = null;
    private int num = 0;
     
    public AbstractHanldMsgThread(KafkaStream<byte[], byte[]> kafkaStream, int num) {
        super();
        this.kafkaStream = kafkaStream;
        this.num = num;
    }
 
    public void run() {
        ConsumerIterator<byte[], byte[]> iterator = kafkaStream.iterator();
        while(iterator.hasNext()) {
            String message = new String(iterator.next().message());
            ExectueMessage(message);
            //System.out.println("Thread no: " + num + ", message: " + message);
        }
    }
    
    public abstract void ExectueMessage(String message);
}
